package com.citusdata.elven.router;

import com.citusdata.elven.router.Endpoint;
import com.citusdata.elven.util.ConfigManager;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;


// Health monitor component that tracks the "client observed" availability of
// each endpoint, and that implements a throttling policy for endpoints that
// appear unavailable.
// More specifically, if the client has problems communicating with an endpoint,
// it registers that failure with this class. If the same endpoint has a certain
// number of failures within a specified time interval (considerFailedInterval),
// this class advises that the endpoint not be contacted until some throttle
// interval elapses.

public class EndpointHealthMonitor {
  private static final Logger logger = Logger.getLogger(EndpointHealthMonitor.class);
  private static final Configuration config = ConfigManager.getConfig();
  
  // The number of past failure records that are retained for each endpoint.
  private static final int MAX_FAILURE_TIMES = 10;
  
  private final int  considerFailedCount;
  private final long considerFailedInterval;  // milliseconds
  private final long throttleInterval;        // milliseconds

  // failureRecords maps an endpoint to a sorted list of past failure times;
  // if the host is to be throttled, throttleTimes maps an endpoint to the time
  // until which the endpoint should not be contacted.
  private ConcurrentMap failureRecords = new ConcurrentHashMap();
  private ConcurrentMap throttleTimes  = new ConcurrentHashMap();


  public EndpointHealthMonitor() {
    String baseKey = "Router.EndpointHealthMonitor.";

    considerFailedCount = config.getInt(baseKey + "ConsiderFailedCount", 3);
    considerFailedInterval = config.getLong(baseKey + "ConsiderFailedInterval", 1000);
    throttleInterval = config.getLong(baseKey + "ThrottleInterval", 1000);

    assert (considerFailedCount <= MAX_FAILURE_TIMES);
  }


  public boolean throttle(Endpoint endpoint, long currentTime) {
    boolean throttle = false;

    Long throttleTime = (Long) throttleTimes.get(endpoint);
    if ((throttleTime != null) && (currentTime <= throttleTime)) {
      throttle = true;
      logger.debug("Throttling connection to endpoint " + endpoint);
    }

    return throttle;
  }


  public void recordFailure(Endpoint endpoint, long failureTime) {
    List failureTimes = failureTimesListGetAndInit(endpoint, failureRecords);

    synchronized(failureTimes) {
      // Insert failureTime to the failureTimes list, and then sort the list by
      // time (from oldest to newest).
      failureTimes.add(failureTime);
      Collections.sort(failureTimes);
      
      // If the failureTimes list exceeded the maximum number of elements it can
      // contain, trim the list.
      if (failureTimes.size() > MAX_FAILURE_TIMES) {
        int oldestIndex = 0;
        failureTimes.remove(oldestIndex);
      }

      // Walk over failureTimes looking for consecutive failures in one "failure
      // interval". If enough consecutive failures are found, calculate the time
      // until which connections will be throttled.
      long throttleTime = -1;
      for (int fi = 0; fi < failureTimes.size() - considerFailedCount + 1; ++fi) {
        long startTime = (Long) failureTimes.get(fi);
        long consecutiveTime = startTime;
        int failureCount = 1;

        for (int ci = fi + 1; ci < fi + considerFailedCount; ++ci) {
          consecutiveTime = (Long) failureTimes.get(ci);
          if (consecutiveTime <= startTime + considerFailedInterval) {
            failureCount++;
          }
        }

        if (failureCount == considerFailedCount) {
          throttleTime = consecutiveTime + throttleInterval;
        }
      }

      // If we calculated a new throttle time, and if that time is further into
      // the future than the previous throttle time, set the calculated throttle
      // time. Also emit an error message if appropriate.
      if (throttleTime >= 0) {
        boolean emitError = false;

        Long prevThrottleTime = (Long) throttleTimes.get(endpoint);
        if (prevThrottleTime == null) {
          throttleTimes.putIfAbsent(endpoint, throttleTime);
          emitError = true;
        } else if (prevThrottleTime < throttleTime) {
          throttleTimes.replace(endpoint, throttleTime);
          emitError = true;
        }

        if (emitError) {
          logger.error("Consecutive failures detected for endpoint " + endpoint + 
                       "; throttling messages until: " + throttleTime + "ms");
        }
      }
    }  // synchronized -- failureTimes
  }


  private List failureTimesListGetAndInit(Endpoint endpoint, 
                                          ConcurrentMap failureRecords) {
 
    // Use the "get then putIfAbsent" idiom. For references, see Effective Java
    // 2nd Edition, Item 69, or Google for "concurrenthashmap example".
    List failureTimes = (List) failureRecords.get(endpoint);
    if (failureTimes == null) {
      List newFailureTimes = new ArrayList();
      failureTimes = (List) failureRecords.putIfAbsent(endpoint, newFailureTimes);
      if (failureTimes == null) {
        failureTimes = newFailureTimes;
      }
    }

    return failureTimes;
  }
}
